#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include <memory>
#include"logger.hpp"
namespace wsll{
class MQClient
{
public:
    using MessageCallback = std::function<void(const char*,size_t)>;
    using ptr = std::shared_ptr<MQClient>;
    MQClient(const std::string &user,const std::string &password,const std::string &host)
    {
        _loop = EV_DEFAULT;
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        std::string url = "amqp://"+user+":"+password+"@"+host+"/";
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), AMQP::Address(url));
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get()); 

        _loop_thread = std::thread([this]() {
            ev_run(_loop, 0);
        });
    }
    void declareComponents(const std::string& exchange,const std::string&queue,const std::string& routing_key = "routing_key",
    AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct)
    {
        _channel->declareExchange(exchange,exchange_type)
        .onError([](const char *message) {
            LOG_ERROR("声明交换机失败:{}", message);
            exit(0);
        })
        .onSuccess([exchange]() {
            LOG_INFO("{}声明交换机成功",exchange);
        });

        _channel->declareQueue(queue)
        .onError([](const char *message) {
            LOG_ERROR("声明队列失败:{}", message);
            exit(0);
        })
        .onSuccess([queue]() {
            LOG_INFO("{}声明队列成功",queue);
        });

        _channel->bindQueue(exchange, queue, routing_key)
        .onError([exchange,queue](const char *message) {
            LOG_ERROR("绑定{}-{}失败,message:{}", exchange,queue,message);
            exit(0);
        })
        .onSuccess([exchange,queue]() {
            LOG_INFO("绑定{}-{}成功",exchange,queue);
        });
    }
    bool publish(const std::string &exchange,const std::string &message,const std::string &routing_key = "routing_key")
    {
        bool ret = _channel->publish(exchange, routing_key, message);
        if(ret)
        {
            LOG_INFO("发布消息成功");
        }
        else
        {
            LOG_ERROR("{}发布消息失败",exchange);
        }
        return ret;
    }
    void consume(const std::string &queue,const MessageCallback& callback)
    {
       
        _channel->consume(queue, "consume-tag")
        .onReceived([this,callback](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
            callback(message.body(),message.bodySize());
            _channel->ack(deliveryTag);
        })
        .onError([queue](const char *message) {
            LOG_ERROR("订阅{}消息失败:{}", queue,message);
            exit(0);
        });
    }
    ~MQClient()
    {
        ev_async_init(&_async_watcher, watcher_callback);//调用c是不能用lambda表达式的
        ev_async_start(_loop, &_async_watcher);
        ev_async_send(_loop, &_async_watcher);
        _loop_thread.join();
        //ev_loop_destroy(_loop);
        _loop = nullptr;
    }
private:
    struct ev_async _async_watcher;
    struct ev_loop *_loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
private:
    static void watcher_callback(struct ev_loop *loop, struct ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);//退出事件循环
    }
};
}